package com.spring.boot.rabbit.service.consumer;

import com.rabbitmq.client.Channel;
import org.slf4j.Logger;
import org.slf4j.LoggerFactory;
import org.springframework.amqp.core.Message;
import org.springframework.amqp.core.MessageProperties;
import org.springframework.amqp.rabbit.annotation.RabbitListener;
import org.springframework.stereotype.Service;

import java.io.IOException;
import java.util.concurrent.ThreadLocalRandom;

import static com.spring.boot.rabbit.constant.QueueConstant.DEFAULT_QUEUE;

/**
 * direct exchange验证多个消费者监听同一个队列
 * <p>
 * 注：多个消费者将轮询处理消息
 *
 * @author 李坤
 * @date 2022/9/1 11:52
 */
@Service
public class RabbitConsumerDirect {

    /**
     * 随机数生成器
     */
    private final ThreadLocalRandom random = ThreadLocalRandom.current();
    /**
     * 日志处理器
     */
    private final Logger logger = LoggerFactory.getLogger(RabbitConsumerDirect.class);

    /**
     * direct exchange binding queue listener one
     *
     * @param body    消息内容
     * @param message 消息具体数据
     * @param channel 消息管道对象
     */
    @RabbitListener(queues = {DEFAULT_QUEUE}, ackMode = "MANUAL")
    public void directExchangeBindingQueueListenerOne(String body, Message message, Channel channel) throws IOException {
        if (consumerMessageConfirm(message, channel)) {
            logger.info("direct exchange queue one: " + body);
        }
    }

    /**
     * direct exchange binding queue listener two
     *
     * @param message 消息队列中的消息
     */
    @RabbitListener(queues = {DEFAULT_QUEUE}, ackMode = "MANUAL")
    public void directExchangeBindingQueueListenerTwo(String body, Message message, Channel channel) throws IOException {
        if (consumerMessageConfirm(message, channel)) {
            logger.info("direct exchange queue two: " + body);
        }
    }

    /**
     * direct exchange binding queue listener three
     *
     * @param message 消息队列中的消息
     */
    @RabbitListener(queues = {DEFAULT_QUEUE}, ackMode = "MANUAL")
    public void directExchangeBindingQueueListenerThree(String body, Message message, Channel channel) throws IOException {
        if (consumerMessageConfirm(message, channel)) {
            logger.info("direct exchange queue three: " + body);
        }
    }


    /*-------------------------------------------------private methods------------------------------------------------*/


    /**
     * 消费端进行手动确认
     *
     * @param message 消息对象
     * @param channel 消息管道
     *
     * @return 是否接收消息
     */
    private boolean consumerMessageConfirm(Message message, Channel channel) throws IOException {
        final MessageProperties properties = message.getMessageProperties();
        final Boolean redelivered = properties.isRedelivered(); // 该消息是否为重新投递
        final long deliveryTag = properties.getDeliveryTag();

        if (random.nextInt(10) % 2 == 0) {
            logger.info("消费端-消息确认: {}；是否为重新投递：{}", true, redelivered);
            channel.basicAck(deliveryTag, true);
            return true;
        } else {
            logger.info("消费端-消息确认: {}；是否为重新投递：{}", false, redelivered);
            channel.basicNack(deliveryTag, false, true);
            return false;
        }
    }

}
